package com.caius.xiaohashu.data.align.consumer;

import com.caius.framework.common.util.JsonUtils;
import com.caius.xiaohashu.data.align.constant.MQConstant;
import com.caius.xiaohashu.data.align.constant.RedisKeyConstant;
import com.caius.xiaohashu.data.align.constant.TableConstants;
import com.caius.xiaohashu.data.align.domain.mapper.InsertMapper;
import com.caius.xiaohashu.data.align.model.dto.CollectUnCollectNoteMqDTO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Component;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.Collections;
import java.util.Objects;

/**
 * @author Caius
 * @description
 * @since Created in 2025-05-29
 */
@Component
@RocketMQMessageListener(consumerGroup = "xiaohashu_group_data_align_" + MQConstant.TOPIC_COUNT_NOTE_COLLECT,
        topic = MQConstant.TOPIC_COUNT_NOTE_COLLECT
)
@Slf4j
public class TodayNoteCollectIncrementData2DBConsumer implements RocketMQListener<String> {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private InsertMapper insertMapper;

    @Value("${table.shards}")
    private int tableShards;

    @Override
    public void onMessage(String body) {
        log.info("## TodayNoteCollectIncrementData2DBConsumer 消费到了 MQ: {}", body);

        CollectUnCollectNoteMqDTO collectUnCollectNoteMqDTO = JsonUtils.parseObject(body, CollectUnCollectNoteMqDTO.class);

        if (Objects.isNull(collectUnCollectNoteMqDTO)) return;

        Long noteId = collectUnCollectNoteMqDTO.getNoteId();
        Long noteCreatorId = collectUnCollectNoteMqDTO.getNoteCreatorId();

        String date = LocalDate.now()
                .format(DateTimeFormatter.ofPattern("yyyyMMdd"));

        // ------------------------- 笔记的收藏数变更记录 -------------------------
        // 笔记对应的 Bloom Key
        String noteBloomKey = RedisKeyConstant.buildBloomUserNoteCollectNoteIdListKey(date);

        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/bloom_today_note_collect_check.lua")));
        script.setResultType(Long.class);

        RedisScript<Long> bloomAddScript = RedisScript.of("return redis.call('BF.ADD', KEYS[1], ARGV[1])", Long.class);

        // 1. 布隆过滤器判断该日增量数据是否已经记录
        Long result = redisTemplate.execute(script, Collections.singletonList(noteBloomKey), noteId);

        if (Objects.equals(result, 0L)) {

            long noteIdHashKey = noteId % tableShards;

            try {
                // - t_data_align_note_collect_count_temp_日期_分片序号
                insertMapper.insert2DataAlignNoteCollectCountTempTable(TableConstants.buildTableNameSuffix(date, noteIdHashKey), noteId);
            } catch (Exception e) {
                log.error("", e);
            }
            //  3. 数据库写入成功后，再添加布隆过滤器中
            redisTemplate.execute(bloomAddScript, Collections.singletonList(noteBloomKey), noteId);
        }

        // ------------------------- 笔记的收藏数变更记录 -------------------------
        // 用户对应的 Bloom Key
        String userBloomKey = RedisKeyConstant.buildBloomUserNoteCollectUserIdListKey(date);

        result = redisTemplate.execute(script, Collections.singletonList(userBloomKey), noteCreatorId);

        if (Objects.equals(result, 0L)) {
            long userIdHashKey = noteCreatorId % tableShards;

            try {
                insertMapper.insert2DataAlignUserCollectCountTempTable(TableConstants.buildTableNameSuffix(date, userIdHashKey), noteCreatorId);
            } catch (Exception e) {
                log.error("", e);
            }

            redisTemplate.execute(bloomAddScript, Collections.singletonList(userBloomKey), noteCreatorId);
        }
    }
}
